package com.neusoft.databus.agent.trap.messaging;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.neusoft.bizcore.common.bean.MessageBean;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.agent.trap.SnmptrapService;
import com.neusoft.databus.common.bean.SnmptrapBean;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
@ConditionalOnProperty(name = "databus.agent.snmpTrap.server.l2CacheProvider", havingValue = "true")
@ConditionalOnClass(KafkaTemplate.class)
public class SnmptrapKafkaConsumer {

    @Autowired
    private SnmptrapService snmptrapService;

    @KafkaListener(group = "${spring.kafka.consumer.group-id}", topics = "${databus.agent.snmpTrap.server.l2CacheTopic}")
    public void receiveTopic(MessageBean message) {
        final String jsonStr = message.getMessage();
        final SnmptrapBean bean = JsonUtils.jsonToPojo(jsonStr, SnmptrapBean.class);
        if (SnmptrapKafkaConsumer.log.isDebugEnabled()) {
            SnmptrapKafkaConsumer.log.debug("receive a snmp trap from kafka {}", bean);
        }
        this.snmptrapService.processTrap(bean);
    }
}
